package com.yootk.rockemq.transaction;

import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.RPCHook;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.util.concurrent.Executor;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class TransactionMessageProducer { // 消息生产者
    public static final String NAME_SERVER_LIST = "rocketmq-server:9876"; // NameServer
    public static final String PRODUCER_GROUP = "muyan-group"; // 定义消费组
    public static final String TOPIC = "TopicTransaction"; // 定义主题名称
    public static final String ACCESS_KEY = "RocketMQMuyan"; // 定义用户名
    public static final String SECRET_KEY = "helloyootk"; // 定义密码

    public static void main(String[] args) throws Exception { // 懒人必备的处理形式
        // 1、由于此时的RocketMQ启动了ACL安全认证的保护机制，所以需要配置相应的回调
        RPCHook clientHook = new AclClientRPCHook(new SessionCredentials(ACCESS_KEY, SECRET_KEY));
        // 2、事务消息创建的核心的主题在于TransactionListener接口实例的配置
        TransactionListener transactionListener = new TransactionListenerImpl();
        // 3、创建事务消息的生产者
        TransactionMQProducer producer = new TransactionMQProducer(PRODUCER_GROUP, clientHook);
        producer.setNamesrvAddr(NAME_SERVER_LIST); // NameServer地址
        producer.setExecutorService(Executors.newFixedThreadPool(8)); // 配置线程池
        producer.setTransactionListener(transactionListener); // 事务监听类
        producer.start(); // 启动生产者
        sendDeptMessage(producer); // 发送部门的消息
        sendEmpMessage(producer); // 发送雇员的消息
        TimeUnit.MINUTES.sleep(Long.MAX_VALUE); // 保持运行状态不关闭
        producer.shutdown(); // 关闭生产者
    }

    public static void sendDeptMessage(TransactionMQProducer producer) throws Exception {
        Message msg = new Message(TOPIC, "dept", "【DEPT】增加新部门事务".getBytes(RemotingHelper.DEFAULT_CHARSET));
        // 发送事务消息的时候需要设置一个业务的标记，同时要返回有具体的事务消息处理的结果
        TransactionSendResult result = producer.sendMessageInTransaction(msg, "DeptProvider");
        TimeUnit.MICROSECONDS.sleep(10); // 模拟分布式的业务调用
    }

    public static void sendEmpMessage(TransactionMQProducer producer) throws Exception {
        for (int x = 0; x < 3; x++) { // 循环配置多个消息
            Message msg = new Message(TOPIC, "emp", ("【EMP】增加新雇员事务 - " + x).getBytes(RemotingHelper.DEFAULT_CHARSET));
            // 发送事务消息的时候需要设置一个业务的标记，同时要返回有具体的事务消息处理的结果
            TransactionSendResult result = producer.sendMessageInTransaction(msg, "EmpProvider");
            TimeUnit.MICROSECONDS.sleep(10); // 模拟分布式的业务调用
        }
    }
}